package com.feidee.data.email.process;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.feidee.data.email.dao.ConfDao;
import com.feidee.data.email.dao.PutDao;
import com.feidee.data.email.dao.QueryDao;
import com.feidee.data.email.model.EConstant;
import com.feidee.data.email.model.EmailLogModel;
import com.feidee.data.email.model.EmailProcessParam;
import com.feidee.data.email.model.EmailQueryParam;
import com.feidee.data.email.model.EmailResult;
import com.feidee.data.email.util.EmailUtil;
import com.feidee.data.email.util.mail.EmailDeal;

@Component
public class EmailProcess {

	private static final Logger LOG = LoggerFactory.getLogger(EmailProcess.class);

	@Resource
	private ConfDao confDao;
	@Resource
	private QueryDao queryDao;
	@Resource
	private PutDao putDao;

	private static final List<String> RUUNING_EMAIL = new ArrayList<String>();

	public EmailResult process(EmailProcessParam processParam, String model) {
		EmailDeal emailDeal = new EmailDeal(processParam.getEmail_key(), model);
		try {
			if (!processParam.isOnly_send()) {
				List<EmailQueryParam> queryParams = confDao.getQueryParam(processParam.getEmail_key());
				if (queryParams.size() == 0) {
					LOG.info(processParam.getEmail_key() + "没有要执行的查询");
					return new EmailResult("-001", processParam.getEmail_key() + "没有要执行的查询");
				}
				boolean invalidate = false;
				for (EmailQueryParam param : queryParams) {
					String time = EmailUtil.getNowTime();
					long start = System.currentTimeMillis();
					EmailLogModel logs = new EmailLogModel();
					logs.setEmail_key(param.getEmail_key());
					logs.setProcess("process");
					logs.setIndex(param.getEmail_index());
					try {
						if ("create".equals(param.getEmail_operation())) {
							queryDao.createWithSql(param.getSource_sql(), param.getSource_engine());

						} else if ("replace".equals(param.getEmail_operation())) {
							if (EConstant.IMPALA.equals(param.getSource_engine()) && !invalidate) {
								LOG.info("invalidate metadata");
								String sql = "invalidate metadata";
								queryDao.createWithSql(sql, EConstant.IMPALA);
								invalidate = true;
							}
							LOG.info("query sql " + param.getEmail_key() + "_" + param.getEmail_index() + ": "
									+ param.getSource_sql());
							List<Map<String, String>> datas = queryDao.queryWithSql(param.getSource_sql(),
									param.getSource_engine());
							String data_check_class = param.getSink_data_check_class();
							Class<?> clazz = Class.forName("default".equals(data_check_class)
									? EConstant.DEFULT_CHECK_CLASS : data_check_class);
							Object o = clazz.newInstance();
							Object check = clazz.getMethod("check", List.class).invoke(o, datas);
							if (!Boolean.parseBoolean(check.toString())) {
								throw new RuntimeException(
										param.getEmail_key() + "邮件 index为" + param.getEmail_index() + " 数据校验失败,请检查源数据");
							}
							LOG.info(param.getEmail_key() + "邮件 index为" + param.getEmail_index() + " 数据校验成功,并写入数据库");
							if ("update".equals(param.getEmail_operation())) {
								putDao.updateDate(datas, param.getSink_table(), param.getSink_keys());
							} else if ("replace".equals(param.getEmail_operation())) {
								String sink_type = param.getSink_type();
								if ("h".equals(sink_type)) {
									putDao.replaceDataWithHorizontal(datas, param.getSink_table());
								} else if ("v".equals(sink_type)) {
									putDao.replaceDateWithVertical(datas, param.getSink_table(), param.getSink_keys());
								}
							}
						}
					} catch (Exception e) {
						logs.setStatus("Failed");
						logs.setMessage(e.getMessage());
						long end = System.currentTimeMillis();
						logs.setCost((int) (end - start));
						logs.setStarttime(time);
						EmailUtil.sendLog(logs, model);
						LOG.info("SQL " + param.getEmail_key() + param.getEmail_index() + "执行失败!", e);
						if (processParam.isSend_email()) {
							try {
								emailDeal.sendErrorMail("SQL执行异常， " + e.getMessage());
								LOG.info("SQL执行异常,发送错误邮件!");
							} catch (Exception e1) {
								e1.printStackTrace();
							}
						}
						return new EmailResult("-001", e.getMessage());
					}
					LOG.info("email_key为" + param.getEmail_key() + "index为" + param.getEmail_index() + "成功写入数据库");
					logs.setStatus("SuccessFully");
					logs.setMessage("None");
					long end = System.currentTimeMillis();
					logs.setCost((int) (end - start));
					logs.setStarttime(time);
					EmailUtil.sendLog(logs, model);
				}
				LOG.info(processParam.getEmail_key() + "所有数据写入数据库,");
			}
			if (processParam.isSend_email()) {
				LOG.info("获取邮件内容并发送");
				EmailResult result = emailDeal.getEmailAndSend();
				if (!"000".equals(result.getCode())) {
					LOG.info("获取邮件及发送失败,发送错误邮件:" + result.getMessage());
					return new EmailResult(result.getCode(), result.getMessage() + "&发送邮件失败,发送错误邮件");
				}
			}
			return new EmailResult("000", "Successfully");
		} catch (Exception e) {
			if (processParam.isSend_email()) {
				try {
					LOG.info("发生异常,发送错误邮件!!");
					emailDeal.sendErrorMail(e.getMessage());
				} catch (Exception e1) {
					e1.printStackTrace();
				}
			}
			return new EmailResult("-001", e.getMessage());
		}
	}

	public void putRunningMail(String email_key) {
		synchronized (RUUNING_EMAIL) {
			RUUNING_EMAIL.add(email_key);
		}

	}

	public boolean isRunning(String email_key) {
		synchronized (RUUNING_EMAIL) {
			return RUUNING_EMAIL.contains(email_key);
		}
	}

	public void removeRunningMail(String email_key) {
		synchronized (RUUNING_EMAIL) {
			RUUNING_EMAIL.remove(email_key);
		}
	}

	public void kill(String email_key) {
		if (isRunning(email_key)) {
			synchronized (RUUNING_EMAIL) {
				RUUNING_EMAIL.remove(email_key);
			}
		}
	}

}
